package club.monkeywood.ad.dmp.tag

import club.monkeywood.ad.dmp.util.{JedisPools, TagsUtils}
import com.typesafe.config.ConfigFactory
import org.apache.hadoop.hbase.{HColumnDescriptor, HTableDescriptor, TableName}
import org.apache.hadoop.hbase.client.{ConnectionFactory, Put}
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapred.TableOutputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.mapred.JobConf
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Row, SparkSession}

import scala.collection.mutable.ListBuffer


object Tags4CtxWithOneUserId extends App {

	if (args.length != 5) {
		println(
			"""
				|cn.dmp.tags.Tags4Ctx
				|参数：
				| 日志输入路径
				| APP字典文件路径
				| 停用词库
				| 日期
				| 输出路径
			""".stripMargin)
		sys.exit()
	}

	val Array(inputPath, dictFilePath, stopWordsFilePath, day, outputPath) = args

	// 2 创建sparkconf->SparkSession
	val sparkConf = new SparkConf()
	sparkConf.setAppName(s"${this.getClass.getSimpleName}")
	sparkConf.setMaster("local[*]")
	//使用KryoSerializer更快
	// RDD 序列化到磁盘 worker与worker之间的数据传输
	sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
	//避免字段名过长报错
	sparkConf.set("spark.debug.maxToStringFields", "100")
	val ss = SparkSession.builder()
		.appName(s"${this.getClass.getSimpleName}")
		.config(sparkConf)
		.getOrCreate()

	// 字典文件---appMapping
	// key:appId
	// value:appName
	val dictMap = ss.sparkContext.textFile(dictFilePath).map(line => {
		val fields = line.split("\t", -1)
		(fields(3), fields(1))
	}).collect().toMap

	// 字典文件 --- stopwords
	val stopWordsMap = ss.sparkContext.textFile(stopWordsFilePath)
												.map((_, 0))  //map比数组性能更好，只用key，不用value,value都取0
												.collect()
												.toMap

	// 将字典数据广播executor
	val broadcastAppDict = ss.sparkContext.broadcast(dictMap)
	val broadcastStopWordsDict = ss.sparkContext.broadcast(stopWordsMap)

	// 将标签信息写入hbase------------------------------------
//	val load = ConfigFactory.load()
//	val hbTableName = load.getString("hbase.table.name")
//
//	// 判断hbase中的表是否存在，如果不存在则创建
//	val configuration = ss.sparkContext.hadoopConfiguration
//	configuration.set("hbase.zookeeper.quorum", load.getString("hbase.zookeeper.host"))
//	configuration.set("zookeeper.znode.parent","/hbase-unsecure")
//	val hbConn = ConnectionFactory.createConnection(configuration)
//	val hbAdmin = hbConn.getAdmin
//
//	if (!hbAdmin.tableExists(TableName.valueOf(hbTableName))) {
//			println(s"$hbTableName 不存在....")
//			println(s"正在创建 $hbTableName ....")
//
//			val tableDescriptor = new HTableDescriptor(TableName.valueOf(hbTableName))
//			//列族名:cf
//			val columnDescriptor = new HColumnDescriptor("cf")
//			tableDescriptor.addFamily(columnDescriptor)
//			hbAdmin.createTable(tableDescriptor)
//
//			// 释放连接
//			hbAdmin.close()
//			hbConn.close()
//	}
//
//	// 指定输出类型
//	val jobConf = new JobConf(configuration)
//	jobConf.setOutputFormat(classOf[TableOutputFormat])
//	// 指定表的名称
//	jobConf.set(TableOutputFormat.OUTPUT_TABLE, hbTableName)


	// 读取日志parquet文件
	// parquet文件中有元数据，可以使用sql语句
	val baseData: RDD[Row] = ss.read.parquet(inputPath)
		.where(TagsUtils.hasSomeUserIdConditition)  //过滤掉没有用户id的数据
		.rdd

  baseData.mapPartitions(par => {

	  val jedis = JedisPools.getConnection()

	  //两种写法：---------------------------------------------------
	  //1）map------------------------------------------------------
	  var tagsOfPar: Iterator[(String, List[(String, Int)])] = par.map(row => {

		  // 行数据进行标签化处理
		  // 广告标签
		  val ads = Tags4Ads.makeTags(row)

		  val apps = Tags4App.makeTags(row, broadcastAppDict.value)
		  //设备标签
		  val devices = Tags4Devices.makeTags(row)
		  //关键字标签
		  val keywords = Tags4KeyWords.makeTags(row, broadcastStopWordsDict.value)
		  //用户id标签
		  //返回的来源于多种渠道的用户id
		  val allUserId: ListBuffer[String] = TagsUtils.getAllUserId(row)

		  // 商圈的标签
		  // 从redis商圈知识库中取经纬度对应的商圈标签
		  val business = Tags4Business.makeTags(row, jedis)

		  //用户id暂时未用，故取第一个id占位即可
		  val userID: String = allUserId(0)
		  val list: List[(String, Int)] = (ads ++ apps ++ devices ++ keywords ++ business).toList
		  //listBuffer.append((userID, list))
		  (userID, list)
	  })
	  jedis.close()
		tagsOfPar

	  //2）foreach------------------------------------------------------------------------
	  //foreach没有返回值，需要创建临时list收集返回值
//		  val listBuffer = new collection.mutable.ListBuffer[(String, List[(String, Int)])]()
//		  //注意：foreach不返回值
//		  par.foreach(row => {
//
//			  // 行数据进行标签化处理
//			  // 广告标签
//			  val ads = Tags4Ads.makeTags(row)
//
//			  val apps = Tags4App.makeTags(row, broadcastAppDict.value)
//			  //设备标签
//			  val devices = Tags4Devices.makeTags(row)
//			  //关键字标签
//			  val keywords = Tags4KeyWords.makeTags(row, broadcastStopWordsDict.value)
//			  //用户id标签
//			  //返回的来源于多种渠道的用户id
//			  val allUserId: ListBuffer[String] = TagsUtils.getAllUserId(row)
//
//			  // 商圈的标签
//			  // 从redis商圈知识库中取经纬度对应的商圈标签
//			  val business = Tags4Business.makeTags(row, jedis)
//
//			  //用户id暂时未用，故取第一个id占位即可
//			  val userID: String = allUserId(0)
//			  val list: List[(String, Int)] = (ads ++ apps ++ devices ++ keywords ++ business).toList
//			  listBuffer.append((userID, list))
//		  })
//		  jedis.close()
//		  listBuffer.iterator

  })
	//聚合相同标签的数量
	.reduceByKey((a, b) => {
		// 第一种写法：使用mapValues聚合
		// (a ++ b)的作用：List(("K电视剧", 1)) ++ List(("APP爱奇艺", 1),("K电视剧", 1)) => List(("K电视剧", 1), ("APP爱奇艺", 1),("K电视剧", 1))
		// groupBy(_._1)的作用：List(("K电视剧", 1), ("APP爱奇艺", 1),("K电视剧", 1)) => Map[ ("K电视剧"，List(("K电视剧", 1), ("K电视剧", 1))), ("APP爱奇艺"，List(("APP爱奇艺", 1)]
		(a ++ b).groupBy(_._1)
			//第1个_表示:List(("K电视剧", 1), ("K电视剧", 1))
			//foldLeft(0)中的0表示：初始值
			//foldLeft(0)(_+_._2)中的第1个_表示：上一次的累加值
			//foldLeft(0)(_+_._2)中的第2个_表示：当前("K电视剧", 1)中的value值=1
			//mapValues输出：Map(("K电视剧",2),("APP爱奇艺",1))
			.mapValues(_.foldLeft(0)(_ + _._2)).toList

		//第二种写法：使用偏函数聚合
		(a ++ b).groupBy(_._1)
			//map后面是{}，表示偏函数
			//偏函数中必须用case匹配输入元素
			.map {
				case (k, sameTags) => (k, sameTags.map(_._2).sum)
		}.toList
	})
//    .map{     //将标签写入hbase
//      case (userId, userTags) => {
//          val put = new Put(Bytes.toBytes(userId))
//          val tags = userTags.map(t => t._1+":"+t._2).mkString(",")
//          //列族：cf
//	        //列名：day20190907
//	        //值：ZP山东省:1,ZC济南市:1,K综艺娱乐:1,K最新更新:1,LC09:1......
//	        put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes(s"day$day"), Bytes.toBytes(tags))
//
//          (new ImmutableBytesWritable(), put)   // ImmutableBytesWritable => rowkey
//      }
//	  }.saveAsHadoopDataset(jobConf)
	//将标签写入文件
	.saveAsTextFile(outputPath)

	ss.close()

}
